package com.mai.realtime.app.dwd.db;

import com.mai.realtime.app.BaseSqlApp;
import com.mai.realtime.common.Constant;
import com.mai.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * com.mai.realtime.app.dwd.db.Dwd_UserRegister
 * 用户注册(新增用户)事实表
 */
public class Dwd_UserRegister extends BaseSqlApp {
    public static void main(String[] args) {
        new Dwd_UserRegister().init(
                3009,
                2,
                "Dwd_UserRegister"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //  1. 读取ods_db
        readOdsDb(tEnv,"Dwd_UserRegister");

        //  2. 读取用户表数据
        Table userInfo = tEnv.sqlQuery("select " +
                "data['id'] user_id, " +
                "date_format(data['create_time'],'yyyy-MM-dd') date_id, " +
                "data['create_time'] create_time, " +
                "ts " +
                "from ods_db "+
                "where `table` = 'user_info' " +
                "and `type` = 'insert' ");
        tEnv.createTemporaryView("user_info", userInfo);

        //  5. 创建 Kafka-Connector dwd_user_register 表
        tEnv.executeSql("create table `dwd_user_register`(" +
                "`user_id` string," +
                "`date_id` string," +
                "`create_time` string," +
                "`ts` bigint " +
                ")"
                + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_USER_REGISTER));

        // 写出
        userInfo.executeInsert("dwd_user_register");
    }
}
